package realtime.app.dws;

import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.types.MapValue;
import org.apache.flink.util.Collector;
import realtime.app.func.BeanToJsonStrFunction;
import realtime.app.func.DimAsyncFunction;
import realtime.app.func.DimJoinFunction;
import realtime.beans.ExamCourseBean;
import realtime.util.DateFormatUtil;
import realtime.util.DorisUtil;
import realtime.util.EnvUtil;
import realtime.util.KafkaUtil;

import java.math.BigDecimal;
import java.util.concurrent.TimeUnit;

public class DwsExamCourseWindow {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = EnvUtil.getSEE(4);
        String topic = "dwd_exam_paper";
        String groupId = "dws_exam_course_window_group";
        KafkaSource<String> kafkaSource = KafkaUtil.getKafkaConsumer(topic, groupId);
        DataStreamSource<String> kafkaStrDS = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka_source");
        //把流中数据转换为JSONObj
        SingleOutputStreamOperator<JSONObject> processDS = kafkaStrDS.process(
                new ProcessFunction<String, JSONObject>() {
                    @Override
                    public void processElement(String jsonStr, ProcessFunction<String, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                        JSONObject jsonObject = JSONObject.parseObject(jsonStr);
                        out.collect(jsonObject);
                    }
                }
        );
        //指定水位线
        SingleOutputStreamOperator<JSONObject> watermarkDS = processDS.assignTimestampsAndWatermarks(
                WatermarkStrategy.<JSONObject>forMonotonousTimestamps()
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<JSONObject>() {
                                    @Override
                                    public long extractTimestamp(JSONObject jsonObject, long l) {
                                        return jsonObject.getLong("ts") * 1000;
                                    }
                                }
                        )
        );
        //按照课程进行分组
        KeyedStream<JSONObject, String> keyedDS = watermarkDS.keyBy(v1 -> v1.getString("course_id"));
        //对流中数据进行处理并且封装为实体类
        SingleOutputStreamOperator<ExamCourseBean> beanDS = keyedDS.process(
                new ProcessFunction<JSONObject, ExamCourseBean>() {
                    private MapState<String, String> userMapState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        MapStateDescriptor<String, String> mapState = new MapStateDescriptor<>("userMapState", String.class, String.class);
                        userMapState = getRuntimeContext().getMapState(mapState);
                    }

                    @Override
                    public void processElement(JSONObject jsonObj, ProcessFunction<JSONObject, ExamCourseBean>.Context ctx, Collector<ExamCourseBean> out) throws Exception {
                        String courseId = jsonObj.getString("course_id");
                        String userId = jsonObj.getString("user_id");
                        BigDecimal score = jsonObj.getBigDecimal("score");
                        Long durationSec = jsonObj.getLong("duration_sec");
                        String userIdState = userMapState.get(userId);
                        Long examUserCt = 0L;
                        if (StringUtils.isEmpty(userIdState)) {
                            examUserCt = 1L;
                            userMapState.put(userId, "1");
                        }
                        out.collect(
                                new ExamCourseBean(
                                        "",
                                        "",
                                        "",
                                        courseId,
                                        "",
                                        examUserCt,
                                        score,
                                        durationSec
                                )
                        );
                    }
                }
        );

        //beanDS.print();
        //按照试卷分组
        KeyedStream<ExamCourseBean, String> courseKeyedDS = beanDS.keyBy(v1 -> v1.getCourseId());
        //开窗
        WindowedStream<ExamCourseBean, String, TimeWindow> windowDS = courseKeyedDS.window(TumblingEventTimeWindows.of(Time.seconds(10)));
        //聚合
        SingleOutputStreamOperator<ExamCourseBean> reduceDS = windowDS.reduce(new ReduceFunction<ExamCourseBean>() {
                                                                                @Override
                                                                                public ExamCourseBean reduce(ExamCourseBean v1, ExamCourseBean v2) throws Exception {
                                                                                    v1.setExamUserCt(v1.getExamUserCt() + v2.getExamUserCt());
                                                                                    v1.setScore(v1.getScore().add(v2.getScore()));
                                                                                    v1.setDurTime(v1.getDurTime() + v2.getDurTime());
                                                                                    return v1;
                                                                                }
                                                                            },
                new WindowFunction<ExamCourseBean, ExamCourseBean, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow window, Iterable<ExamCourseBean> input, Collector<ExamCourseBean> out) throws Exception {
                        String stt = DateFormatUtil.toYmdHms(window.getStart());
                        String edt = DateFormatUtil.toYmdHms(window.getEnd());
                        long timeMillis = System.currentTimeMillis();
                        String curDate = DateFormatUtil.toDate(timeMillis);
                        for (ExamCourseBean examCourseBean : input) {
                            examCourseBean.setStt(stt);
                            examCourseBean.setEdt(edt);
                            examCourseBean.setCurDate(curDate);
                            out.collect(examCourseBean);
                        }
                    }
                }
        );
        //补充维度字段
        SingleOutputStreamOperator<ExamCourseBean> resultDS = AsyncDataStream.unorderedWait(
                reduceDS,
                new DimAsyncFunction<ExamCourseBean>("dim_course_info") {
                    @Override
                    public String getKey(ExamCourseBean bean) {
                        return bean.getCourseId();
                    }

                    @Override
                    public void join(ExamCourseBean bean, JSONObject dim) {
                        bean.setCourseName(dim.getString("course_name"));
                    }
                },
                60,
                TimeUnit.SECONDS
        );
        //resultDS.print(">>>>");
        resultDS.map(new BeanToJsonStrFunction<>())
                .sinkTo(DorisUtil.getDorisSink("dws_exam_course_window"));

        env.execute();
    }
}
